/**
 * 
 */
package com.ccb.channel.p8;

import java.io.IOException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.alibaba.fastjson.JSON;
import com.ccb.util.Record;
import com.ccb.util.RecordType;
import com.ccb.util.StandardRecord;

//import com.alibaba.fastjson.JSON;

/**
 * @author congyang
 *
 */
public class P8Reducer extends Reducer<Text, Text, Text, NullWritable> {
	String reqXml = null;
	String resXml = null;
	String trace_id = null;
	String service_code = null;
	String[] keys = null;
	String resJson = null;
	String reqJson = null;
	Text output = new Text();
	// static List<String> jsonList = new ArrayList<>();
	private MultipleOutputs<Text, NullWritable> mos = null;

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		mos = new MultipleOutputs<Text, NullWritable>(context);
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		mos.close();
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		StandardRecord standardRecord = new StandardRecord();
		for (Text val : values) {
			Record rd = JSON.parseObject(val.toString(), Record.class);
			standardRecord.setUid(rd.getUid());
			standardRecord.setServiceId(rd.getServiceId());
			if (rd.getRecordType().equals(RecordType.REQUEST)) {
				standardRecord.setRequestData(rd.getData());
				standardRecord.setRequest(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.REQUEST);
				standardRecord.setTime(rd.getTime());

			}else if (rd.getRecordType().equals(RecordType.RESPONSE)) {
				standardRecord.setResponseData(rd.getData());
				standardRecord.setResponse(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.RESPONSE);
				standardRecord.setResponseCode(rd.getResponseCode());
			}
		}
		if (standardRecord.getRequest() != null && standardRecord.getResponse() != null)
			standardRecord.setRecordType(RecordType.BOTH);
		String service_code = standardRecord.getServiceId();
		System.out.println(service_code);
		mos.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get(), service_code);
		// context.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get());
	}

}
